/*
 * =====================================================================================
 *       Filename:  circle.c
 *         Author:  MIEN
 *    Description:  生产者/消费者模式的环形队列
 *                  使用示例参考 src/dispatcher.c 和 test/circle.c
 *           Date:  2016-11-16
 * =====================================================================================
 */

#include <stdlib.h>
#include <stdio.h>
#include "circle.h"

circleQueue* circleCreate(void)
{
    circleQueue *circle;

    if ((circle = malloc(sizeof(circleQueue))) == NULL)
       return NULL;

    circle->head = NULL;
    circle->tail = NULL;
    circle->full = NULL;

    circle->capacity = 0;
    circle->tasks = 0;

    circle->free = NULL;
    circle->produce = NULL;
    circle->consume = NULL;

    circle->ct = NULL;
    circle->loop = 1;
 
    pthread_mutex_init(&(circle->lock), NULL);
    pthread_cond_init(&(circle->sig), NULL);

    return circle;
}

void circleRelease(circleQueue *circle)
{
    unsigned long len;
    circleNode *current, *next;

    current = circle->head;
    len = circle->capacity;
    while(len--) {
        next = current->next;
        if (circle->free) circle->free(current->value);
        free(current);
        current = next;
    }

    circleThread *n, *t = circle->ct;
    while (t != NULL) {
        n = t;
        t = t->next;
        if (circle->free) circle->free(n->value);
        free(n);
    }

    pthread_cond_destroy(&(circle->sig));
    pthread_mutex_destroy(&(circle->lock));

    free(circle);
}

void circlePut(circleQueue *circle, void* data)
{
    pthread_mutex_lock(&(circle->lock));
    
    if (circle->capacity - circle->tasks == 0) {
        circleNode* n = (circleNode*) malloc(sizeof(circleNode));
        n->value =  NULL;

        if (circle->head == NULL) {
            circle->head = n;
            circle->tail = n;
            circle->full = n;
            n->next = n;
        } else {
            /* 此时head所指向的节点也是有任务数据的 */
            n->next = circle->head;
            circle->tail->next = n;
            circle->tail = n;
        }

        circle->capacity++;
    } else {
        circle->tail = circle->head;
        circle->head = circle->head->next;
    }

    /* 处理外部传入的data数据，设置节点的value值 */
    circle->produce(data, &(circle->tail->value));

    circle->tasks++;
    if (circle->tasks == 1) {
        pthread_cond_signal(&(circle->sig));
    }

    pthread_mutex_unlock(&(circle->lock));
}

static void* circleGet(void *ct)
{
    circleThread *thread = (circleThread*) ct;
    circleQueue *circle = thread->circle;

    while (1) {

        pthread_mutex_lock(&(circle->lock));

        while (circle->tasks == 0) {
            if (circle->loop == 0) {
                pthread_mutex_unlock(&(circle->lock));
                return NULL;
            }
            pthread_cond_wait(&(circle->sig), &(circle->lock));
        }

        /* 
         * 交换:
         * full节点的value <=> thread节点的value
         * 只需交换指针数值，不必挪动数据实体。
         */
        void *v = circle->full->value;
        circle->full->value = thread->value;
        thread->value = v;

        circle->full = circle->full->next;
        circle->tasks--;

        pthread_mutex_unlock(&(circle->lock));

        /* 调用指定的外部 consume 方法 */
        circle->consume(thread->value);
    }
}

void circleAddThread(circleQueue *circle)
{
    circleThread *n = malloc(sizeof(circleThread));
    n->next = NULL;
    n->proc = circleGet;
    circle->produce(NULL, &(n->value));
    n->circle = circle;

    if (circle->ct == NULL) {
        circle->ct = n;
    } else {
        circleThread *t = circle->ct;
        while (t != NULL) {
            if (t->next == NULL) {
                t->next = n;
                break;
            }
            t = t->next;
        }
    }
}

void circleLoop(circleQueue *circle)
{
    circle->loop = 1;
    circleThread *t = circle->ct;
    while (t != NULL) {
        pthread_create(&(t->id), NULL, t->proc, t);
        t = t->next;
    }
}

void circleQuit(circleQueue *circle)
{
    circle->loop = 0;
    circleThread *t = circle->ct;
    pthread_cond_broadcast(&(circle->sig));
    while (t != NULL) {
        pthread_join(t->id, NULL);
        t = t->next;
    }
}



